package com.lhz.collector.source;

import org.apache.flume.Event;
import org.apache.flume.event.EventBuilder;
import org.jboss.netty.buffer.ChannelBuffer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.ByteArrayOutputStream;
import java.util.HashMap;
import java.util.Map;

public class NettyUtils {
  private ByteArrayOutputStream baos;
  private static final Logger logger = LoggerFactory.getLogger(NettyUtils.class);
  public static final Integer MIN_SIZE = 10;
  public static final Integer DEFAULT_SIZE = 2500;
  private final boolean isUdp;
  private Integer maxSize;

  public NettyUtils() {
    this(false);
  }

  public NettyUtils(boolean isUdp) {
    this(DEFAULT_SIZE, isUdp);
  }

  public NettyUtils(Integer eventSize, boolean isUdp) {
    this.isUdp = isUdp;
    maxSize = (eventSize < MIN_SIZE) ? MIN_SIZE : eventSize;
    baos = new ByteArrayOutputStream(eventSize);
  }

  // create the event from syslog data
  Event buildEvent() {
    try {
      byte[] body=baos.toByteArray();
      Map<String, String> headers = new HashMap<String, String>();
      // format the message
      return EventBuilder.withBody(body, headers);
    } finally {
      reset();
    }
  }

  private void reset() {
    baos.reset();
  }

  public Event extractEvent(ChannelBuffer in) {
    byte b = 0;
    Event e = null;
    boolean doneReading = false;

    try {
      while (!doneReading && in.readable()) {
        b = in.readByte();
        // TCP syslog entries are separated by '\n'
        if (b == '\n') {
          e = buildEvent();
          doneReading = true;
        } else {
          baos.write(b);
        }
        if (baos.size() == this.maxSize && !doneReading) {
          e = buildEvent();
          doneReading = true;
        }
      }
      // UDP doesn't send a newline, so just use what we received
      if (e == null && isUdp) {
        doneReading = true;
        e = buildEvent();
      }
    } finally {
      // no-op
    }

    return e;
  }

  public Integer getEventSize() {
    return maxSize;
  }

  public void setEventSize(Integer eventSize) {
    this.maxSize = eventSize;
  }

}


